package io.kimmking.rpcfx.client;

import com.alibaba.fastjson.JSON;
import io.kimmking.rpcfx.context.RpcfxContext;
import io.kimmking.rpcfx.tcp.CustomTcpDecoder;
import io.kimmking.rpcfx.tcp.CustomTcpEncoder;
import io.kimmking.rpcfx.tcp.CustomTcpMessage;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author lfy
 */
@Slf4j
public class NettyClient {
    private static long timeout=3;
    private static TimeUnit timeUnit = TimeUnit.SECONDS;
    private static int times = 3;

    public void setTimeoutParam(long timeout,TimeUnit timeUnit,int times){
        NettyClient.timeout = timeout;
        NettyClient.timeUnit = timeUnit;
        NettyClient.times = times;
    }

    private static ConcurrentHashMap<String, Channel> channelPool = new ConcurrentHashMap<>();

    public static String getResponse(String url, String message)  {
        for (int i = 0; i < times; i++) {
            try {
                return getResponseCall(url,message);
            }catch (Exception ex){
                System.out.println("调用出现异常,次数:"+(i+1)+",异常信息:"+ex.getMessage());
            }
        }
        throw new RuntimeException("调用失败,超出最大重试次数");
    }
    public static String getResponseCall(String url, String message) throws InterruptedException,
            URISyntaxException {
        URI uri = new URI(url);
        String cacheKey = uri.getHost() + ":" + uri.getPort();
        CustomTcpMessage request = buildRequest(uri, message);
        // 查看缓存池中是否有可重用的channel
        if (channelPool.containsKey(cacheKey)) {
            Channel channel = channelPool.get(cacheKey);
            if (!channel.isActive() || !channel.isWritable() || !channel.isOpen()) {
                log.debug("通道不可用，即将重新创建");
                channelPool.remove(cacheKey);
            } else {
                return handler(request, channel);
            }
        }
        // 没有或者不可用则新建
        // 并将最终的handler添加到pipeline中，拿到结果后返回
        Channel channel = createChannel(uri.getHost(), uri.getPort());
        String result = handler(request, channel);
        channelPool.put(cacheKey, channel);
        return result;
    }

    private static String handler(CustomTcpMessage request, Channel channel) throws InterruptedException {
        //
        synchronized (channel) {
            HttpClientAsyncHandler handler = new HttpClientAsyncHandler();
            channel.pipeline().replace("clientHandler", "clientHandler", handler);
            channel.writeAndFlush(request).sync();
            return handler.getResponse(timeout,timeUnit);
        }
    }

    public static CustomTcpMessage buildRequest(URI uri, String message) throws URISyntaxException {
        CustomTcpMessage request = new CustomTcpMessage();
        //header
        request.setHeader(JSON.toJSONString(RpcfxContext.getAll()));
        request.setBody(message);
        return request;
    }

    /**
     * 返回新的Channel
     *
     * @param address ip地址
     * @param port    端口
     * @return channel
     * @throws InterruptedException exception
     */
    private static Channel createChannel(String address, int port) throws InterruptedException {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(workerGroup);
        b.channel(NioSocketChannel.class);
        b.option(ChannelOption.SO_REUSEADDR, true);
        b.option(ChannelOption.TCP_NODELAY, true);
        b.option(ChannelOption.AUTO_CLOSE, true);
        b.option(ChannelOption.SO_KEEPALIVE, true);
        b.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new CustomTcpEncoder());
                ch.pipeline().addLast(new CustomTcpDecoder());
                ch.pipeline().addLast("clientHandler", new HttpClientAsyncHandler());
            }
        });
        return b.connect(address, port).sync().channel();
    }
}
